package com.xxpt.common.utils.amqp.util;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xxpt.common.utils.amqp.QueueCode;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;

/**
 * rabbitMQ自定义工具类
 * @author icetea
 */
@Slf4j
public class RabbitMqUtil {
    public static final String CHARSET_NAME = "UTF-8";
    //最大重试次数
    private static final int MAX_RETRY_TIMES = 3;
    //重试次数在消息头中的head的key值
    private static final String RETRY_HEAD_KEY = "x-retry-time";
    //默认死信交换器
    public static final String DEFAULT_EXCHANGE_DLX =  "xxpt.dlx"; // 死信交换器
    //服务默认topic交换机
    public static String EXCHANGE_TOPIC = "xxpt.topic";
    //服务默认直连交换机
    public static final String EXCHANGE_DIRECT = "xxpt.direct";
    //服务默认延迟直连交换机
    public static final String EXCHANGE_DELAYED_DIRECT = "xxpt.delayed.direct";
    //异常栈保留最字符数
    private static final int MAX_STACK_TRACE_LENGTH = Integer.MAX_VALUE;

    private RabbitTemplate rabbitTemplate;
    private RabbitTemplate rabbitTemplateDelay;

    private RabbitMqUtil() {

    }

    public RabbitMqUtil(RabbitTemplate rabbitTemplate, RabbitTemplate rabbitTemplateDelay) {
        this.rabbitTemplate = rabbitTemplate;
        this.rabbitTemplateDelay = rabbitTemplateDelay;
    }

    /**
     * 发送json数据
     * @param exchangeName 交换器名称
     * @param routingKey   路由键
     * @param obj          消息javaBean
     */
    public void sendMessage(String exchangeName, String routingKey, Object obj) {
        rabbitTemplate.convertAndSend(exchangeName, routingKey, obj);
        log.info("发送消息==>exchangeName={},routingKey={},{}", exchangeName, routingKey, JSON.toJSONString(obj));
    }

    /**
     * 根据队列枚举发送消息
     * @param queue 队列枚举类
     * @param obj    消息实体
     */
    public void sendMessage(QueueCode queue, Object obj) {
        rabbitTemplate.convertAndSend(EXCHANGE_TOPIC, queue.getName(), obj);
        log.info("发送消息==>queue={},{}", queue.getName(), JSON.toJSONString(obj));
    }

    /**
     * 发送延迟消息
     * @param queue
     * @param obj
     */
    public void sendDelayedMessage(QueueCode queue, Object obj) {
        sendDelayedMessage(queue, obj, MessageProperties.CONTENT_TYPE_JSON);
    }

    /**
     * 发送延迟消息
     * @param queue
     * @param obj
     */
    public void sendDelayedMessage(QueueCode queue, Object obj, String contentType) {
        Message msg = null;
        try {
            msg = MessageBuilder.withBody(JSON.toJSONString(obj).getBytes("UTF-8"))
                    .setHeader("x-delay", queue.getDelayTime() * 1000)
                    .setContentType(StringUtils.isNotBlank(contentType) ? contentType : "")
                    .build();
        } catch (Exception ex) {
            log.error("RabbitMqUtil.sendDelayedMessage.fail=>{}", obj, ex);
            return;
        }
        //设置 Mandatory=true 时消息无法路由。报错：NO_ROUTE
        rabbitTemplateDelay.convertAndSend(EXCHANGE_DELAYED_DIRECT, queue.getRoutingKey(), msg);
        log.info("发送延时消息==>queue={},{}", queue.getName(), JSON.toJSONString(obj));
    }

    /**
     * 消费者处理消息方法，自动重新入队方法
     * @param message          消息（消息头+消息体）
     * @param jsonObject       消息体
     * @param dlxRoutingKey    死信队列路由键（一般为死信队列名称）
     * @param messageProcessor 消息处理器
     */
    public void handleMessageWithRetry(Message message, Channel channel, JSONObject jsonObject, String dlxRoutingKey, MessageProcessor messageProcessor) {
        MessageProperties messageProperties = message.getMessageProperties();
        Map<String, Object> headers = messageProperties.getHeaders();// 获取自定义头信息
        try {
            messageProcessor.processMessage(jsonObject);
            //确认
            channel.basicAck(messageProperties.getDeliveryTag(),false);
        } catch (Exception e) {
            // 处理出现异常,进行重试
            int retryTimes = headers.containsKey(RETRY_HEAD_KEY) ? (int) headers.get(RETRY_HEAD_KEY) : 0;
            int nextRetryTimes = retryTimes + 1;
            if (nextRetryTimes > MAX_RETRY_TIMES) {
                // 超出最大尝试次数，将其放置到死信队列,并保存错误信息
                this.rabbitTemplate.convertAndSend(DEFAULT_EXCHANGE_DLX, dlxRoutingKey, message,
                        (Message msg) -> {
                            // 记录异常信息
                            msg.getMessageProperties().setHeader("x-exception-message", e.getCause() != null ? e.getCause().getMessage() : e.getMessage());
                            // 记录异常栈
                            msg.getMessageProperties().setHeader("x-exception-stacktrace", getStackTraceString(e));
                            // 记录异常入队时间
                            msg.getMessageProperties().setHeader("x-exception-time", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
                            return msg;
                        });
                log.error("message={} 已超过最大重试次数，放置到死信队列中", new String(message.getBody()));
            } else {
                try {
                    //手动确认回滚 拒绝deliveryTag对应的消息，第二个参数是否requeue，true则重新入队列，否则丢弃或者进入死信队列。
                    channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
                } catch (IOException ioException) {
                    ioException.printStackTrace();
                }
            }
            /*else {
                // 入队重试
                this.rabbitTemplate.convertAndSend(messageProperties.getReceivedExchange(), messageProperties.getConsumerQueue(),
                        message, (Message msg) -> {
                            // 更新重试次数
                            msg.getMessageProperties().setHeader(RETRY_HEAD_KEY, nextRetryTimes);
                            return msg;
                        });
                log.info("message={}处理失败，进行第 {} 次重试", new String(message.getBody()), nextRetryTimes);
            }*/
        }
    }

    /**
     * 处理消息接口
     */
    public interface MessageProcessor {
        void processMessage(JSONObject jsonObject);
    }

    /**
     * copy RepublishMessageRecoverer中的代码
     * @param cause 异常
     * @return 根据最大保留字符数返回异常栈信息
     */
    private String getStackTraceString(Throwable cause) {
        StringWriter stringWriter = new StringWriter();
        PrintWriter printWriter = new PrintWriter(stringWriter, true);
        cause.printStackTrace(printWriter);
        String stackTraceAsString = stringWriter.getBuffer().toString();

        if (stackTraceAsString.length() > MAX_STACK_TRACE_LENGTH) {
            stackTraceAsString = stackTraceAsString.substring(0, MAX_STACK_TRACE_LENGTH);
        }
        printWriter.close();

        return stackTraceAsString;
    }
}
